package com.gitee.ywj1352.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.stereotype.Component;

import java.util.Properties;

/**
 * @author wenjie.yang on 2019/6/21.
 */
@Configuration
@Component
public class KafkaConsumerConfig {

//    @Bean
//    public ConsumerFactory<Integer, String> consumerFactory() {
//        return new DefaultKafkaConsumerFactory(consumerConfigs());
//    }
//
//    @Bean
//    public Properties consumerConfigs() {
//        Properties properties = new Properties();
//        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
//        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"KafkaProducerDemo");
//        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
//        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");
//        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.IntegerDeserializer");
//        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
//        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
//        return properties;
//    }
//
//    @Bean
//    ConcurrentKafkaListenerContainerFactory<Integer, String> getKafkaListenerContainerFactory(ConsumerFactory consumerFactory) {
//        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =new ConcurrentKafkaListenerContainerFactory<>();
//        factory.setConsumerFactory(consumerFactory);
//        factory.setConcurrency(1);
//        factory.setBatchListener(true);
//        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
//        return factory;
//    }


}
